package ip
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object IPAddress {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("IPAddress").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")

    /*//读取文件
    val http: RDD[String] = sc.textFile("data/http.log")
    val ip: RDD[String] = sc.textFile("data/ip.dat")

    //处理http数据
    val httpRDD: RDD[(String,String)] = http.map {
      line =>
        val field = line.split("\\|")
        (field(1),line)
    }
    //println(httpRDD.count())

    //处理ip数据
    val ipRDD: RDD[((String, String), String)] = ip.map {
      line =>
        val field = line.split("\\|")
        ((field(0), field(1)), field(7))
    }

    //求出log中ip所属城市

    val httpRDDInfo: RDD[(String,String)] = httpRDD.map(key => key).distinct()


    def getIplong(ip: String): Long = {
      val ips = ip.split("\\.")
      var ip2long = 0L
      for (i <- 0 until 4) {
        ip2long = ip2long << 8 | Integer.parseInt(ips(i))
      }
      ip2long
    }

    def whichCity(ip:String,startIp:String,endIp:String):Boolean = {
      getIplong(startIp) <= getIplong(ip)&& getIplong(ip) <= getIplong(endIp)
    }


    val ip_addr: RDD[(String, String)] = ipRDD.cartesian(httpRDDInfo).
      filter {
        case (((startIp, endIp), _), (ip, _)) => whichCity(ip, startIp, endIp)
      }.map {
      case (((_, _), addr), (ip, _)) => (ip, addr)
    }
*/
    // 读数据，并解析。将ip地址转换为Long类型
    val httpData: RDD[Long] = sc.textFile("data/http.log")
      .map(x => ip2Long(x.split("\\|")(1)))

    // 读数据，解析，收回来。最后变为广播变量
    val ipData: Array[(Long, Long, String)] = sc.textFile("data/ip.dat")
      .map { line =>
        val field = line.split("\\|")
        (field(2).toLong, field(3).toLong, field(6))
      }.collect()
    // 此处对ipData进行排序后再广播，方便后面使用二分法查找城市
    val ipBC = sc.broadcast(ipData.sortBy(_._1))

    // 逐条数据比对，找到对应的城市。使用二分查找
    httpData.mapPartitions { iter =>
      val ipsInfo: Array[(Long, Long, String)] = ipBC.value
      iter.map { ip =>
        val city: String = getCityName(ip, ipsInfo)
        (city, 1)
      }
    }.reduceByKey(_ + _)
      .collect()
      .foreach(println)

    sc.stop()
  }

  // 将ip地址转换为Long。有多种实现方法，此方法使用了位运算，效率最高
  def ip2Long(ip: String): Long = {
    ip.split("\\.")
      .map(_.toLong)
      .fold(0L) { (buffer, elem) =>
        buffer << 8 | elem
      }
  }

  // 给定ip地址，在ips中查找对应的城市名。使用二分查找算法
  def getCityName(ip: Long, ips: Array[(Long, Long, String)]): String = {
    var start = 0
    var end = ips.length - 1
    var middle = 0

    while (start <= end) {
      middle = (start + end) / 2
      if ((ip >= ips(middle)._1) && (ip <= ips(middle)._2))
        return ips(middle)._3
      else if (ip < ips(middle)._1)
        end = middle - 1
      else
        start = middle + 1
    }
    "Unknown"
  }









    //val ipaddrBc: Broadcast[collection.Map[String, String]] = sc.broadcast(ip_addr)
//ipaddrBc.value.foreach(println)
    /*val value: RDD[(String, String)] = httpRDD.map {
      case (ip, line) =>
        val ipaddrInfo: collection.Map[String, String] = ipaddrBc.value
        val str: String = ipaddrInfo.getOrElse(ip, null)
        (str, line)
    }*/


    //ipRDD.foreach(println)









}
